/*
 * Copyright (C) 2013, 2014 Brett Wooldridge
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.zaxxer.hikari.util;

import static com.zaxxer.hikari.util.ClockSource.currentTime;
import static com.zaxxer.hikari.util.ClockSource.elapsedNanos;
import static com.zaxxer.hikari.util.ConcurrentBag.IConcurrentBagEntry.STATE_IN_USE;
import static com.zaxxer.hikari.util.ConcurrentBag.IConcurrentBagEntry.STATE_NOT_IN_USE;
import static com.zaxxer.hikari.util.ConcurrentBag.IConcurrentBagEntry.STATE_REMOVED;
import static com.zaxxer.hikari.util.ConcurrentBag.IConcurrentBagEntry.STATE_RESERVED;
import static java.lang.Thread.yield;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
import static java.util.concurrent.locks.LockSupport.parkNanos;

import java.lang.ref.WeakReference;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.zaxxer.hikari.util.ConcurrentBag.IConcurrentBagEntry;

/**
 * HikariCP连接池是基于自主实现的ConcurrentBag完成的数据连接的多线程共享交互，是HikariCP连接管理快速的其中一个关键点。
 * <br/>ConcurrentBag是一个专门的并发包裹，在连接池（多线程数据交互）的实现上具有比LinkedBlockingQueue和LinkedTransferQueue更优越的性能。
 * <br/>ConcurrentBag通过拆分CopyOnWriteArrayList、ThreadLocal和SynchronousQueue进行并发数据交互。
 * <br/>CopyOnWriteArrayList：负责存放ConcurrentBag中全部用于出借的资源
 * <br/>ThreadLocal：保存当前线程用过的连接，用于加速线程本地化资源访问
 * <br/>SynchronousQueue：所有可以使用的资源都会offer到这里，无缓存等待连接的队列，一手交钱一手交货
 * <br/>This is a specialized concurrent bag
 * that achieves superior performance to LinkedBlockingQueue and
 * LinkedTransferQueue for the purposes of a connection pool. It uses
 * ThreadLocal storage when possible to avoid locks, but resorts to scanning a
 * common collection if there are no available items in the ThreadLocal list.
 * Not-in-use items in the ThreadLocal lists can be "stolen" when the borrowing
 * thread has none of its own. It is a "lock-less" implementation using a
 * specialized AbstractQueuedLongSynchronizer to manage cross-thread signaling.
 *
 * Note that items that are "borrowed" from the bag are not actually removed
 * from any collection, so garbage collection will not occur even if the
 * reference is abandoned. Thus care must be taken to "requite" borrowed objects
 * otherwise a memory leak will result. Only the "remove" method can completely
 * remove an object from the bag.
 *
 * @author Brett Wooldridge
 *
 * @param <T> the templated type to store in the bag
 */
public class ConcurrentBag<T extends IConcurrentBagEntry> implements AutoCloseable {
	private static final Logger LOGGER = LoggerFactory.getLogger(ConcurrentBag.class);

	private final CopyOnWriteArrayList<T> sharedList;
	private final boolean weakThreadLocals;

	private final ThreadLocal<List<Object>> threadList;
	private final IBagStateListener listener;
	private final AtomicInteger waiters;
	private volatile boolean closed;

	private final SynchronousQueue<T> handoffQueue;

	public interface IConcurrentBagEntry {
		int STATE_NOT_IN_USE = 0;
		int STATE_IN_USE = 1;
		int STATE_REMOVED = -1;
		int STATE_RESERVED = -2;

		boolean compareAndSet(int expectState, int newState);

		void setState(int newState);

		int getState();
	}

	public interface IBagStateListener {
		void addBagItem(int waiting);
	}

	/**
	 * Construct a ConcurrentBag with the specified listener.
	 *
	 * @param listener the IBagStateListener to attach to this bag
	 */
	public ConcurrentBag(final IBagStateListener listener) {
		this.listener = listener;
		this.weakThreadLocals = useWeakThreadLocals();

		this.handoffQueue = new SynchronousQueue<>(true);
		this.waiters = new AtomicInteger();
		this.sharedList = new CopyOnWriteArrayList<>();
		if (weakThreadLocals) {
			this.threadList = ThreadLocal.withInitial(() -> new ArrayList<>(16));
		} else {
			this.threadList = ThreadLocal.withInitial(() -> new FastList<>(IConcurrentBagEntry.class, 16));
		}
	}

	/**ConcurrentBag中通过borrow方法进行数据资源借用，通过requite方法进行资源回收，注意其中borrow方法只提供对象引用，不移除对象，因此使用时通过borrow取出的对象必须通过requite方法进行放回，否则容易导致内存泄露
	 * <br/>The method will borrow a BagEntry from the bag, blocking for the specified timeout if none are available.
	 *
	 * @param timeout  how long to wait before giving up, in units of unit
	 * @param timeUnit a <code>TimeUnit</code> determining how to interpret the timeout parameter
	 * @return a borrowed instance from the bag or null if a timeout occurs
	 * @throws InterruptedException if interrupted while waiting
	 */
	public T borrow(long timeout, final TimeUnit timeUnit) throws InterruptedException {
		// 优先查看有没有可用的本地化的资源
		// 细节就是倒着遍历，最后面的是最近用过的，说明刚还的，很可能还没被别的线程抢过去
		final List<Object> list = threadList.get();
		for (int i = list.size() - 1; i >= 0; i--) {
			// 每遍历到一个元素都会从thradlocal的list里删除，然后尝试状态从0-1的修改，如果修改不成功说明状态是1,-1,-2，那就不用管了，继续拿下个。本质上就是把不能用的资源从本地删除了
			final Object entry = list.remove(i);
			@SuppressWarnings("unchecked")
			final T bagEntry = weakThreadLocals ? ((WeakReference<T>) entry).get() : (T) entry;
			// 如果本地化的资源有，并且可以把PoolEntry状态从STATE_NOT_IN_USE改为STATE_IN_USE，则直接返回该bagEntry
			if (bagEntry != null && bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
				return bagEntry;
			}
		}

		// 如果本地找不到，等待队列+1
		final int waiting = waiters.incrementAndGet();
		try {
			// 当无可用本地化资源时，遍历全部资源，查看是否存在可用资源，因此被一个线程本地化的资源也可能被另一个线程“抢走”
			// 窃取的概念：本来属于某个线程的本地连接，当它归还连接的时，恰巧有另一个线程从sharedList遍历找到这个连接，这时候连接的状态是STATE_NOT_IN_USE，那么这个连接就会被另一个线程也保存到ThreadLocal中了
			for (T bagEntry : sharedList) {
				if (bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
					// If we may have stolen another waiter's connection, request another bag add.
					if (waiting > 1) {
						// 因为可能“抢走”了其他线程的资源，因此提醒包裹进行资源添加，这个listener是异步增加连接用的
						listener.addBagItem(waiting - 1);
					}
					return bagEntry;
				}
			}

			// 走到这里说明现有的资源都在使用中
			// 如果没有获取到，就异步生成waiting个连接
			listener.addBagItem(waiting);

			timeout = timeUnit.toNanos(timeout);
			do {
				final long start = currentTime();
				// 当现有全部资源全部在使用中，等待一个被释放的资源或者一个新资源
				// 获取并删除该队列的头部，如果有必要，将等待到指定的等待时间，以等待另一个线程插入它
				final T bagEntry = handoffQueue.poll(timeout, NANOSECONDS);
				if (bagEntry == null || bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_IN_USE)) {
					return bagEntry;
				}

				timeout -= elapsedNanos(start);
			} while (timeout > 10_000);

			return null;
		} finally {
			// 等待队列-1
			waiters.decrementAndGet();
		}
	}

	/**这个方法将向包返回一个借来的对象。从包中借来但从不“归还”的对象将导致内存泄漏
	 * <br/>This method will return a borrowed object to the bag. Objects that are
	 * borrowed from the bag but never "requited" will result in a memory leak.
	 *
	 * @param bagEntry the value to return to the bag
	 * @throws NullPointerException  if value is null
	 * @throws IllegalStateException if the bagEntry was not borrowed from the bag
	 */
	public void requite(final T bagEntry) {
		// 将状态转为未在使用，因为只有当前线程持有bagEntry，所以可以直接设置
		bagEntry.setState(STATE_NOT_IN_USE);

		// 判断是否存在等待线程，若存在，则直接转手资源，做到及时唤醒
		for (int i = 0; waiters.get() > 0; i++) {
			if (bagEntry.getState() != STATE_NOT_IN_USE || handoffQueue.offer(bagEntry)) {
				return;
			} else if ((i & 0xff) == 0xff) {
				// 每循环256次判断连接是否被拿走，如果没有被拿走，就睡10ns
				parkNanos(MICROSECONDS.toNanos(10));
			} else {
				// 让出时间片
				yield();
			}
		}

		// 否则，进行资源本地化保存
		final List<Object> threadLocalList = threadList.get();
		threadLocalList.add(weakThreadLocals ? new WeakReference<>(bagEntry) : bagEntry);
	}

	/**ConcurrentBag中全部的资源均只能通过add方法进行添加，只能通过remove方法进行移出
	 * <br/>Add a new object to the bag for others to borrow.
	 *
	 * @param bagEntry an object to add to the bag
	 */
	public void add(final T bagEntry) {
		if (closed) {
			LOGGER.info("ConcurrentBag has been closed, ignoring add()");
			throw new IllegalStateException("ConcurrentBag has been closed, ignoring add()");
		}
		// 新添加的资源优先放入CopyOnWriteArrayList
		sharedList.add(bagEntry);

		// 当有等待资源的线程时，将资源交到某个等待线程后才返回（SynchronousQueue）
		while (waiters.get() > 0 && !handoffQueue.offer(bagEntry)) {
			yield();
		}
	}

	/**修改PoolEntry状态，从shareList和threadList中移除这个元素
	 * <br/>Remove a value from the bag. This method should only be called with objects
	 * obtained by <code>borrow(long, TimeUnit)</code> or <code>reserve(T)</code>
	 *
	 * @param bagEntry the value to remove
	 * @return true if the entry was removed, false otherwise
	 * @throws IllegalStateException if an attempt is made to remove an object from
	 *                               the bag that was not borrowed or reserved first
	 */
	public boolean remove(final T bagEntry) {
		// 如果资源正在使用且无法进行状态切换，则返回失败
		if (!bagEntry.compareAndSet(STATE_IN_USE, STATE_REMOVED)
				&& !bagEntry.compareAndSet(STATE_RESERVED, STATE_REMOVED) && !closed) {
			LOGGER.warn("Attempt to remove an object from the bag that was not borrowed or reserved: {}", bagEntry);
			return false;
		}

		final boolean removed = sharedList.remove(bagEntry);// 从CopyOnWriteArrayList中移出
		if (!removed && !closed) {
			LOGGER.warn("Attempt to remove an object from the bag that does not exist: {}", bagEntry);
		}

		return removed;
	}

	/**关闭bag
	 * <br/>Close the bag to further adds.
	 */
	@Override
	public void close() {
		closed = true;
	}

	/**返回指定状态的PoolEntry集合快照
	 * <br/>This method provides a "snapshot" in time of the BagEntry items in the bag in
	 * the specified state. It does not "lock" or reserve items in any way. Call
	 * <code>reserve(T)</code> on items in list before performing any action on
	 * them.
	 *
	 * @param state one of the {@link IConcurrentBagEntry} states
	 * @return a possibly empty list of objects having the state specified
	 */
	public List<T> values(final int state) {
		final List<T> list = sharedList.stream().filter(e -> e.getState() == state).collect(Collectors.toList());
		Collections.reverse(list);
		return list;
	}

	/**
	 * This method provides a "snapshot" in time of the bag items. It does not
	 * "lock" or reserve items in any way. Call <code>reserve(T)</code> on items in
	 * the list, or understand the concurrency implications of modifying items,
	 * before performing any action on them.
	 *
	 * @return a possibly empty list of (all) bag items
	 */
	@SuppressWarnings("unchecked")
	public List<T> values() {
		return (List<T>) sharedList.clone();
	}

	/**这种方法是用来使袋子里的物品“不可借”。它主要用于想要操作由values(int)方法返回的项。被保留的物品可以通过remove(T)从袋子中取出，而不需要取消保留。没有从包中取出的物品可以通过调用unreserve(T)方法再次使用
	 * <br/>The method is used to make an item in the bag "unavailable" for borrowing. It
	 * is primarily used when wanting to operate on items returned by the
	 * <code>values(int)</code> method. Items that are reserved can be removed from
	 * the bag via <code>remove(T)</code> without the need to unreserve them. Items
	 * that are not removed from the bag can be make available for borrowing again
	 * by calling the <code>unreserve(T)</code> method.
	 *
	 * @param bagEntry the item to reserve
	 * @return true if the item was able to be reserved, false otherwise
	 */
	public boolean reserve(final T bagEntry) {
		// 修改PoolEntry的state字段值：从STATE_NOT_IN_USE到STATE_RESERVED
		return bagEntry.compareAndSet(STATE_NOT_IN_USE, STATE_RESERVED);
	}

	/**
	 * This method is used to make an item reserved via <code>reserve(T)</code>
	 * available again for borrowing.
	 *
	 * @param bagEntry the item to unreserve
	 */
	public void unreserve(final T bagEntry) {
		if (bagEntry.compareAndSet(STATE_RESERVED, STATE_NOT_IN_USE)) {
			// spin until a thread takes it or none are waiting
			while (waiters.get() > 0 && !handoffQueue.offer(bagEntry)) {
				yield();
			}
		} else {
			LOGGER.warn("Attempt to relinquish an object to the bag that was not reserved: {}", bagEntry);
		}
	}

	/**获取挂起(等待)某item从bag中变为可用的线程数
	 * <br/>Get the number of threads pending (waiting) for an item from the bag to become available.
	 *
	 * @return the number of threads waiting for items from the bag
	 */
	public int getWaitingThreadCount() {
		return waiters.get();
	}

	/**获取在此调用时处于指定状态的项数量的计数
	 * <br/>Get a count of the number of items in the specified state at the time of this call.
	 *
	 * @param state the state of the items to count
	 * @return a count of how many items in the bag are in the specified state
	 */
	public int getCount(final int state) {
		int count = 0;
		for (IConcurrentBagEntry e : sharedList) {
			if (e.getState() == state) {
				count++;
			}
		}
		return count;
	}

	public int[] getStateCounts() {
		final int[] states = new int[6];
		for (IConcurrentBagEntry e : sharedList) {
			++states[e.getState()];
		}
		states[4] = sharedList.size();
		states[5] = waiters.get();

		return states;
	}

	/**
	 * Get the total number of items in the bag.
	 *
	 * @return the number of items in the bag
	 */
	public int size() {
		return sharedList.size();
	}

	public void dumpState() {
		sharedList.forEach(entry -> LOGGER.info(entry.toString()));
	}

	/**根据类和系统类加载器之间是否存在自定义类加载器实现，确定是否使用弱引用。
	 * <br/>Determine whether to use WeakReferences based on whether there is a custom ClassLoader implementation sitting between this class and the System ClassLoader.
	 *
	 * @return true if we should use WeakReferences in our ThreadLocals, false otherwise
	 */
	private boolean useWeakThreadLocals() {
		try {
			// 人工指定是否使用弱引用，但是官方不推荐进行自主设置
			if (System.getProperty("com.zaxxer.hikari.useWeakReferences") != null) { // undocumented manual override of WeakReference behavior
				return Boolean.getBoolean("com.zaxxer.hikari.useWeakReferences");
			}

			// 默认通过判断初始化的ClassLoader是否是系统的ClassLoader来确定
			return getClass().getClassLoader() != ClassLoader.getSystemClassLoader();
		} catch (SecurityException se) {
			return true;
		}
	}
}
